﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;
using System.Xml.Schema;
using Apache.NMS;
using Apache.NMS.Util;

namespace Common
{
    public delegate void MessageReceivedHandler(object sender, MessageReceivedEventArgs e);

    public class MessageReceivedEventArgs : EventArgs
    {
        private XDocument message;
        private IPrimitiveMap properties;

        public XDocument Message
        {
            get { return message; }
        }

        public IPrimitiveMap Properties
        {
            get { return properties; }
        }

        //public MessageReceivedEventArgs(XDocument message)
        //{
        //    this.message = message;
        //}

        public MessageReceivedEventArgs(XDocument message, IPrimitiveMap properties)
        {
            this.message = message;
            this.properties = properties;
        }
    }

    /// <summary>
    /// http://activemq.apache.org/nms/nms-simple-asynchronous-consumer-example.html
    /// </summary>
    public class QueueHandler
    {
        private string listenerQueueName;
        private XmlSchemaSet messageSchema;
        private IConnectionFactory factory;
        protected AutoResetEvent semaphore = new AutoResetEvent(false);

        private bool abort = false;

        public event MessageReceivedHandler MessageReceived;

        public QueueHandler(string listnerQueueName, XmlSchemaSet messageSchema)
        {
            this.listenerQueueName = listnerQueueName;
            this.messageSchema = messageSchema;

            // NOTE: ensure the nmsprovider-activemq.config file exists in the executable folder.
            factory = new NMSConnectionFactory(Constants.QUEUE_URI);
        }

        public void SendMessage(string sendQueueName, string message, IPrimitiveMap properties)
        {
            using (IConnection connection = factory.CreateConnection())
            using (ISession session = connection.CreateSession())
            {
                IDestination destination = SessionUtil.GetDestination(session, sendQueueName);
                using (IMessageProducer producer = session.CreateProducer(destination))
                {
                    producer.DeliveryMode = MsgDeliveryMode.Persistent;
                    ITextMessage request = session.CreateTextMessage(message);
                    FillProperies(request, properties);
                    producer.Send(request);
                }
            }
        }

        private void FillProperies(ITextMessage request, IPrimitiveMap properties)
        {
            //MULE_CORRELATION_ID	
            //MULE_ROOT_MESSAGE_ID
            //MULE_CORRELATION_SEQUENCE	
            //MULE_CORRELATION_GROUP_SIZE	

            if (properties == null)
                return;

            bool ok = properties.Contains("MULE_CORRELATION_ID");
            ok = ok && properties.Contains("MULE_ROOT_MESSAGE_ID");
            ok = ok && properties.Contains("MULE_CORRELATION_SEQUENCE");
            ok = ok && properties.Contains("MULE_CORRELATION_GROUP_SIZE");

            if(!ok)
                return;

            request.Properties.SetString("MULE_CORRELATION_ID", properties["MULE_CORRELATION_ID"].ToString());
            request.Properties.SetString("MULE_ROOT_MESSAGE_ID", properties["MULE_ROOT_MESSAGE_ID"].ToString());
            request.Properties.SetString("MULE_CORRELATION_SEQUENCE", properties["MULE_CORRELATION_SEQUENCE"].ToString());
            request.Properties.SetString("MULE_CORRELATION_GROUP_SIZE", properties["MULE_CORRELATION_GROUP_SIZE"].ToString());
        }

        public void StartListner()
        {
            if (abort)
                return;

            Task runner = new Task(() => 
            {
                using (IConnection connection = factory.CreateConnection())
                using (ISession session = connection.CreateSession())
                {
                    IDestination destination = SessionUtil.GetDestination(session, listenerQueueName);
                    // Create a consumer
                    using (IMessageConsumer consumer = session.CreateConsumer(destination))
                    {
                        connection.Start();
                        consumer.Listener += consumer_Listener;
                        while (!abort)
                        {
                            semaphore.WaitOne();
                        }
                    }
                }
            });

            runner.Start();
        }

        void consumer_Listener(IMessage message)
        {
            var properties = message.Properties;
            var m  = message as ITextMessage;
 
            string strXml = m.Text;

            bool error = false;
            string errorText = "";

            try
            {
                //Validate XML velformed
                XDocument doc = XDocument.Parse(strXml);

                //Schema validate XML
                doc.Validate(messageSchema, (o, e) =>
                {
                    error = true;
                    errorText = e.Message;
                }, true);
                if (error)
                {
                    doc = null;
                }

                OnMessageReceived(new MessageReceivedEventArgs(doc, properties));

            }
            catch (Exception e)
            {
                throw new Exception(e.Message);
            }
            finally
            {
                semaphore.Set();
            }
        }

        public void StopListener()
        {
            if (abort)
                return;

            abort = true;
            semaphore.Set();
        }

        protected virtual void OnMessageReceived(MessageReceivedEventArgs e)
        {
            if (MessageReceived != null)
                MessageReceived(this, e);
        }

    }
}